package com.mango.ch03;

import java.io.IOException;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopN_CatDemoReducer extends Reducer<NullWritable, Text, NullWritable, Text> {
	private SortedMap<Double, String> finaTop10cats = new TreeMap<Double, String>();
	private Integer N = 10;

	@Override
	protected void reduce(NullWritable key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		// 收集所有本地的top10元素列表
		for (Text catRecord : values) {
			// 重新从整行数据里挑出想要的字段
			String[] tokens = catRecord.toString().split(",");
			Double weight = Double.valueOf(tokens[0]);
			finaTop10cats.put(weight, catRecord.toString());
			// 只保留N个元素 删除最小体重的元素
			if (finaTop10cats.size() > N)
				finaTop10cats.remove(finaTop10cats.firstKey());
		}
		// 完成最终收集后 发出所有元素
		for (String text : finaTop10cats.values()) {
			System.out.println("in reducer---> key="+NullWritable.get()+"  value="+text);
			context.write(NullWritable.get(), new Text(text));
		}
	}

	@Override
	protected void setup(Reducer<NullWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration conf = context.getConfiguration();
		this.N = Integer.valueOf(conf.get("top.n"));// 获取配置文件中设置的N的限制数
	}

}
